package com.dongzili.common.disruptor.handler;

import com.dongzili.common.disruptor.entity.MessageEvent;
import com.dongzili.common.disruptor.service.DisruptorMqService;
import com.dongzili.common.disruptor.service.MessageProcessor;
import com.dongzili.common.redis.RedisOperationsUtil;
import com.lmax.disruptor.EventHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.stereotype.Component;

import jakarta.annotation.Resource;


@Slf4j
@Component
public class MessageEventHandler<T> implements EventHandler<MessageEvent<T>> {

    private static final HashOperations<String, String, MessageEvent<Object>> hashOperations = RedisOperationsUtil.getHashOperations();

    @Resource
    MessageProcessor<T> processor;

    @Override
    public void onEvent(MessageEvent<T> event, long sequence, boolean endOfBatch) {

        try {
            //这里停止1000ms是为了确定消费消息是异步的
//            log.info("sequence:{} 消费者处理消息开始",sequence);
            if (event != null) {
                event.setSequence(sequence);
//                log.info("sequence:{} 消费者消费的信息是：{}",sequence, event);
                hashOperations.delete(DisruptorMqService.redisKey, event.getId());
//                MessageProcessor processor = SpringUtil.getBean(MessageProcessor.class);
                processor.process(event.getData().getClass(),event);
            }
        } catch (Exception e) {
            log.error("sequence:{} 消费者处理消息失败: {}",sequence, e.getMessage(), e);
        }
//        log.info("sequence:{} 消费者处理消息结束", sequence);
    }

}